package org.example.jasper

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions._

object BZnumber {
  case class Result(DistanceRange: String, AccidentCount: Long)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("BZ")
    val sc = new SparkContext(conf)
    // 创建 SparkSession
    val spark = SparkSession.builder
      .appName("BZnumber")
      .getOrCreate()

    // 读取数据
    val inputPath = "src/main/java/org/datas/bz/acidentes2013.csv"
    val df = spark.read
      .option("header", "true")
      .csv(inputPath)

    // 将 "km" 列转换为数字类型
    val dfWithKm = df.withColumn("km", col("km").cast("double"))

    // 定义行驶距离的区间
    val distanceBins = Array(0, 10, 100, 200, 300, 400, 500, 1000)
    val distanceColumn = col("km")

    // 使用 expr 和 case 创建分箱列
    val distanceRangeColumn = expr(
      s"""
         |CASE
         |  WHEN $distanceColumn IS NULL THEN 'Unknown'
         |  WHEN $distanceColumn >= 0 AND $distanceColumn <= 10 THEN '0-10'
         |  WHEN $distanceColumn > 10 AND $distanceColumn <= 100 THEN '11-100'
         |  WHEN $distanceColumn > 100 AND $distanceColumn <= 200 THEN '101-200'
         |  WHEN $distanceColumn > 200 AND $distanceColumn <= 300 THEN '201-300'
         |  WHEN $distanceColumn > 300 AND $distanceColumn <= 400 THEN '301-400'
         |  WHEN $distanceColumn > 400 AND $distanceColumn <= 500 THEN '401-500'
         |  WHEN $distanceColumn > 500 AND $distanceColumn <= 1000 THEN '500-1000'
         |  ELSE 'Unknown'
         |END as DistanceRange
      """.stripMargin
    )

    // 统计每个区间的车祸数量
    val resultDF = dfWithKm
      .groupBy(distanceRangeColumn.alias("DistanceRange"))
      .agg(count("*").alias("AccidentCount"))
      .orderBy("DistanceRange")

    // 打印结果
    resultDF.show()

    // 将结果导入到 MySQL 数据库中
    val mysqlUrl = "jdbc:mysql://localhost:3306/BZaccident"
    val mysqlProperties = new java.util.Properties()
    mysqlProperties.setProperty("user", "root")
    mysqlProperties.setProperty("password", "011216")
    mysqlProperties.setProperty("driver", "com.mysql.jdbc.Driver")

    resultDF.write
      .mode(SaveMode.Overwrite) // 如果表已存在，覆盖之前的数据
      .jdbc(mysqlUrl, "BZnumber", mysqlProperties)
  }
}
